package com.shujia.flink.state;

import org.apache.flink.api.common.functions.RuntimeContext;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.runtime.state.hashmap.HashMapStateBackend;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.environment.CheckpointConfig;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.util.Collector;

import java.util.HashMap;

public class Demo1ValueState {
    public static void main(String[] args) throws Exception {
        //1、创建flink的执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        /*
         * checkpoint作用：定时将flink计算的状态持久化到HDFS中，保存任务失败重启后，状态不丢失
         */
        //开启checkpoint
        //env.enableCheckpointing(5000);

        // 使用 externalized checkpoints，这样 checkpoint 在作业取消后仍就会被保留
        //env.getCheckpointConfig().setExternalizedCheckpointCleanup( CheckpointConfig.ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION);

        //指定状态保存的位置
        //env.setStateBackend(new HashMapStateBackend());

        //checkpoint保存数据的位置
        //env.getCheckpointConfig().setCheckpointStorage("hdfs://master:9000/flink/checkpoint");


        //nc -lk 8888
        DataStream<String> linesDS = env.socketTextStream("master", 8888);

        KeyedStream<String, String> keyByDS = linesDS.keyBy(word -> word);

        //统计单词的数量
        DataStream<Tuple2<String, Integer>> countDS = keyByDS
                .process(new KeyedProcessFunction<String, String, Tuple2<String, Integer>>() {
                    //状态和普通java变量的区别：状态中保存的结果会被checkpoint持久化到hdfs中，任务执行失败状态不会丢失
                    //ValueState单值状态，为每一个key在状态中保存一个值
                    ValueState<Integer> countState;

                    /**
                     * open方法每一个task执行一次，在open方法中进行初始化操作，比如初始化状态
                     */
                    @Override
                    public void open(Configuration parameters) throws Exception {
                        //获取flink的环境
                        RuntimeContext context = getRuntimeContext();

                        //创建状态的描述对象
                        ValueStateDescriptor<Integer> stateDescriptor = new ValueStateDescriptor<>("count", Types.INT);

                        //创建状态
                        //状态和普通java变量的区别：状态中保存的结果会被checkpoint持久化到hdfs中，任务执行失败状态不会丢失
                        countState = context.getState(stateDescriptor);
                    }

                    @Override
                    public void processElement(String word,//一行数据
                                               KeyedProcessFunction<String, String, Tuple2<String, Integer>>.Context ctx, //上下文对象
                                               Collector<Tuple2<String, Integer>> out//用于将数据发送到下游
                    ) throws Exception {
                        //获取状态中保存到结果
                        Integer count = countState.value();

                        //状态中的初始值默认是null
                        if (count == null) {
                            count = 0;
                        }

                        //累加计算
                        count++;

                        //更新状态的值
                        countState.update(count);

                        //将计算结果发送到下游
                        out.collect(Tuple2.of(word, count));
                    }
                });

        countDS.print();

        env.execute();

    }
}
